import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.bean.TradeSkuOrderBean
import com.cw.realtime.common.constant.Constant.{DWS_TRADE_SKU_ORDER_WINDOW, TOPIC_DWD_TRADE_ORDER_DETAIL}
import com.cw.realtime.common.util.{Cache4Dim, FlinkSinkUtil}
import com.cw.realtime.common.util.TimeWindowUtil.WinSyntax
import io.circe._
import io.circe.generic.auto._
import io.circe.syntax._
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{MapState, MapStateDescriptor, StateTtlConfig}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration
import scala.util.Try

object DwsTradeSkuOrderWindow {


  def main(args: Array[String]): Unit = {
    implicit val conf = RunnerConfig("dws_trade_sku_order_window", TOPIC_DWD_TRADE_ORDER_DETAIL, 10027, ckp = true, parallelism = 1)
    run { (_, ds) =>
      //      ds.print()
      // 过滤
      val jsonObjectStream = etl(ds)
      //      jsonObjectStream.map(_.toJson.noSpaces).print()
      // 水位线
      val jsonObjectStreamWM = withWatermark(jsonObjectStream)

      // 修复回撤流的重复数据， 并转化成指标class
      val skuOrderStream = fixRetraction(jsonObjectStreamWM)

      // 开窗汇总
      val aggStream = windowAggregate(skuOrderStream)
            aggStream.print()

      // 维度关联.
      val associatedStream = dimAssociation(aggStream)
      associatedStream.print()
      // sink
            associatedStream
              .map(_.asJson.noSpaces)
              .sinkTo(FlinkSinkUtil.getDorisSink(DWS_TRADE_SKU_ORDER_WINDOW))


    }


  }


  def dimAssociation(ds: DataStream[TradeSkuOrderBean]) = {
    ds.map(new RichMapFunction[TradeSkuOrderBean, TradeSkuOrderBean] {
      var cache = Option.empty[Cache4Dim]

      override def open(parameters: Configuration): Unit = {
        cache = Some(new Cache4Dim)
      }

      override def close(): Unit = cache.foreach(_.close())

      override def map(value: TradeSkuOrderBean): TradeSkuOrderBean = cache.fold(value) { cache =>
        val skuInfo = cache.get("dim_sku_info", value.sku_id)
        //            println(skuInfo)
        val category3_id = skuInfo("category3_id")
        val trademark_id = skuInfo("tm_id")
        val spu_id = skuInfo("spu_id")
        val sku_name = skuInfo("sku_name")

        val spuInfo = cache.get("dim_spu_info", spu_id)
        val spu_name = spuInfo("spu_name")

        val dimCat3 = cache.get("dim_base_category3", category3_id)
        val category3_name = dimCat3("name")
        val category2_id = dimCat3("category2_id")

        val dimCat2 = cache.get("dim_base_category2", category2_id)
        val category2_name = dimCat2("name")
        val category1_id = dimCat2("category1_id")

        val dimCat1 = cache.get("dim_base_category1", category1_id)
        val category1_name = dimCat1("name")

        val trademark = cache.get("dim_base_trademark", trademark_id)
        val trademark_name = trademark("tm_name")

        TradeSkuOrderBean(value.order_detail_id, value.sku_id, value.original_amount, value.activity_reduce_amount, value.coupon_reduce_amount, value.order_amount, value.ts, trademark_id, trademark_name, category1_id, category1_name, category2_id, category2_name, category3_id, category3_name, sku_name, spu_id, spu_name, value.stt, value.edt, value.cur_date)
      }

    })
  }


  def windowAggregate(ds: DataStream[TradeSkuOrderBean]) = {
    ds.keyBy(_.sku_id)
      .windowAll(TumblingEventTimeWindows.of(time.Time.seconds(10)))
      .reduce(_ + _, new AllWindowFunction[TradeSkuOrderBean, TradeSkuOrderBean, TimeWindow] {
        override def apply(window: TimeWindow, input: Iterable[TradeSkuOrderBean], out: Collector[TradeSkuOrderBean]): Unit = {
          val (stt, edt, cur_date) = window.windowRange
          input.foreach { e => out.collect(e.transWindowRange(stt, edt, cur_date)) }
        }
      })
  }


  def fixRetraction(ds: DataStream[JsonObject]) = {
    ds.keyBy(_("id").get.asString.get)
      .process(new KeyedProcessFunction[String, JsonObject, TradeSkuOrderBean] {

        type FuncT = KeyedProcessFunction[String, JsonObject, TradeSkuOrderBean]

        val desc = new MapStateDescriptor("last_amount", classOf[String], classOf[BigDecimal])
        var state = Option.empty[MapState[String, BigDecimal]]

        override def open(parameters: Configuration): Unit = {
          desc.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(3)).build)
          state = Try(getRuntimeContext.getMapState(desc)).toOption
        }

        override def processElement(value: JsonObject, ctx: FuncT#Context, out: Collector[TradeSkuOrderBean]): Unit = {
          for {
            amountState <- state
            ts <- value("ts").get.asNumber.get.toLong
            skuId <- value("sku_id")
            skuId <- skuId.asString
            id <- value("id")
            id <- id.asString
            orderAmount <- value("split_total_amount")
            orderAmount <- orderAmount.asString
            activityAmount <- value("split_activity_amount")
            activityAmount <- activityAmount.asString
            couponAmount <- value("split_coupon_amount")
            couponAmount <- couponAmount.asString
            skuNum <- value("sku_num")
            skuNum <- skuNum.asString
            skuPrice <- value("order_price")
            skuPrice <- skuPrice.asString
          } {
            //优惠前的总金额
            //            if (amountState.get("original_amount") != null)
            //              println("回撤流 order_id:" + id)
            val lastOriginalAmount = Option(amountState.get("original_amount")).getOrElse(BigDecimal(0))
            val lastActivityAmount = Option(amountState.get("activity_reduce_amount")).getOrElse(BigDecimal(0))
            val lastCouponAmount = Option(amountState.get("coupon_reduce_amount")).getOrElse(BigDecimal(0))
            val lastOrderAmount = Option(amountState.get("order_amount")).getOrElse(BigDecimal(0))
            val originalAmount = BigDecimal(skuNum) * BigDecimal(skuPrice)
            out.collect(
              TradeSkuOrderBean(
                id,
                skuId,
                originalAmount - lastOriginalAmount,
                BigDecimal(activityAmount) - lastActivityAmount,
                BigDecimal(couponAmount) - lastCouponAmount,
                BigDecimal(orderAmount) - lastOrderAmount,
                ts
              )
            )
            amountState.put("original_amount", originalAmount)
            amountState.put("activity_reduce_amount", BigDecimal(activityAmount))
            amountState.put("coupon_reduce_amount", BigDecimal(couponAmount))
            amountState.put("order_amount", BigDecimal(orderAmount))

          }

        }
      })
  }


  def withWatermark(ds: DataStream[JsonObject]) = {
    ds.assignTimestampsAndWatermarks(
      WatermarkStrategy.forBoundedOutOfOrderness
      (Duration.ofSeconds(3)).withTimestampAssigner(
        new SerializableTimestampAssigner[JsonObject] {
          override def extractTimestamp(element: JsonObject, recordTimestamp: Long): Long =
            element("ts").get.asNumber.get.toLong.get
        }))
  }


  def etl(ds: DataStream[String]) = {
    ds.flatMap { s =>
      for {
        json <- parser.parse(s).toOption // 过滤掉上游回撤流产生的-D数据
        jsonObj <- json.asObject
        ts <- jsonObj("ts")
        _ <- jsonObj("id")
        _ <- jsonObj("sku_id")
        ts <- ts.asNumber
        ts <- ts.toLong
      } yield
        jsonObj.add("ts", (ts * 1000).asJson)
    }
  }


}
